package fun.tan90.device.sink.mqtt.plugin.handler;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import fun.tan90.device.sink.core.SinkHandler;
import fun.tan90.device.sink.core.SinkRule;
import fun.tan90.device.sink.mqtt.plugin.config.MqttClientTemplate;
import fun.tan90.device.sink.mqtt.plugin.model.SinkMqttRule;
import lombok.extern.slf4j.Slf4j;
import org.dromara.mica.mqtt.codec.MqttQoS;
import org.noear.snack.ONode;
import org.noear.solon.annotation.Component;

import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Slf4j
@Component
public class SinkMqttHandler extends SinkHandler {

    @Override
    public void init(List<ONode> onodeList) {
        for (ONode node : onodeList) {
            String deviceType = node.get("deviceType").getString();
            String dataSource = node.get("dataSource").getString();
            ONode sinkContext = ONode.loadStr(node.get("sinkContext").getString());
            List<SinkMqttRule> sinkRules = sinkContext.ary().stream().map(n -> JSONUtil.toBean(n.toJson(), SinkMqttRule.class)).collect(Collectors.toList());
            SINK_RULES.put(sinkRuleKey(deviceType, dataSource), sinkRules);
        }
    }

    @Override
    public void handle(Map<String, Object> data, Map<String, String> meta, List<? extends SinkRule> sinkRules) {
        for (SinkRule sinkRule : sinkRules) {
            SinkMqttRule sinkMqttRule = (SinkMqttRule) sinkRule;
            String topic = StrUtil.format(sinkMqttRule.getTopic(), data);
            MqttClientTemplate mqttClientTemplate = getAppContext().getBean(sinkMqttRule.getName());
            boolean publish = mqttClientTemplate.publish(topic, ONode.load(data).toJson(), MqttQoS.valueOf(sinkMqttRule.getQos()), sinkMqttRule.isRetain());
            log.info("publish: {}, topic: {}", publish, topic);
        }
    }
}
